package com.shujia.flink.window

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, ProcessingTimeSessionWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object Demo3SessionWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //读取卡口过车数据
    val dataDS: DataStream[String] = env.socketTextStream("master", 8888)

    //整理数据取出道路编号和时间戳
    val kcDS: DataStream[(String, Long)] = dataDS.map(line => {
      val split: Array[String] = line.split(",")
      //道路编号
      val roadId: String = split(1)
      //时间戳
      val ts: Long = split(2).toLong
      (roadId, ts)
    })

    //设置时间字段
    val assDS: DataStream[(String, Long)] = kcDS.assignAscendingTimestamps(_._2)


    val kvDS: DataStream[(String, Int)] = assDS.map(kv => (kv._1, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    /**
     * 会话窗口 -- 同一个key一段时间没有数据开始计算
     * ProcessingTimeSessionWindows： 处理时间的会话窗口
     * EventTimeSessionWindows： 事件时间的会话窗口，需要指定时间字段
     *
     *
     */

    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = keyByDS
      .window(EventTimeSessionWindows.withGap(Time.seconds(5)))

    val countDS: DataStream[(String, Int)] = windowDS.sum(1)

    countDS.print()

    env.execute()

  }

}
